package org.spark.example

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

/**
  * @Author: chengjj
  * @Date: 2020-09-15
  */
object StreamingKafka {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[*]").setAppName("WorldCount1")
        conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
        val ssc = new StreamingContext(conf, Seconds(3))
        ssc.checkpoint("E:\\test\\spark\\checkpoint")
        val preferredHosts = LocationStrategies.PreferConsistent
        val kafkaParams = Map[String, Object](
            "bootstrap.servers" -> "localhost:9092",
            "key.deserializer" -> classOf[StringDeserializer],
            "value.deserializer" -> classOf[StringDeserializer],
            "group.id" -> "test",
            "auto.offset.reset" -> "latest",
            "enable.auto.commit" -> (false: java.lang.Boolean)
        )

        val topics = Array("test")
        val srouceStream = KafkaUtils.createDirectStream[String, String](
            ssc,
            preferredHosts,
            Subscribe[String, String](topics, kafkaParams)
        )



        srouceStream.flatMap(_.value().split(" "))
            .map((_,1))
            /*.reduceByKey(_+_)
            .updateStateByKey[Int]((seq:Seq[Int],opt:Option[Int]) => {
                seq.foreach(println)
                Some(seq.sum + opt.getOrElse(0))
                })*/
                .reduceByKeyAndWindow((x,y)=>x+y,(x, y)=> x-y, Seconds(9), Seconds(6))
                    .filter(s => s._2 > 0)
            .print(10000)
        val offsetRanges = srouceStream.asInstanceOf[HasOffsetRanges].offsetRanges
        // 确保结果都已经正确且幂等地输出了
        srouceStream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
        ssc.start()
        ssc.awaitTermination()
    }
}
